package org.ngbx.demo.flink.dd.statistics;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.ngbx.demo.flink.dd.DeviceDataEvent;

import java.util.Properties;

public class DeviceDataEventStatisticsJob {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建kafka配置属性
        Properties kafkaProps = createKafkaProperties();

        KafkaSource<DeviceDataEvent> source = KafkaSource.<DeviceDataEvent>builder()
                .setProperties(kafkaProps)
                .setTopics("device_data_event")
                .setGroupId("device_data_event_statistics")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new DeviceDataEventDeserializationSchema())
                .build();

        DataStream<DeviceDataEvent> deviceDataEventDS = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Device Event Source");

        deviceDataEventDS.keyBy(DeviceDataEvent::getId)
                .process(new KeyedProcessFunction<String, DeviceDataEvent, String>() {
                    private ValueState<DeviceState> deviceState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 必须在 open 生命周期初始化
                        deviceState = getRuntimeContext().getState(new ValueStateDescriptor<>("deviceState", DeviceState.class));
                    }

                    @Override
                    public void processElement(DeviceDataEvent event, KeyedProcessFunction<String, DeviceDataEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        // 获取最新设备状态
                        DeviceState lastedDeviceState = deviceState.value();

                        Integer deviceStatus = getDeviceStatus(event);

                        if (deviceStatus == null) {
                            out.collect("异常告警数据" + event);
                        } else {
                            if (lastedDeviceState == null) {
                                deviceState.update(new DeviceState(event.getId(), deviceStatus, event.getOccurrenceTime()));
                                System.out.println("初始化设备状态" + deviceState.value().toString());
                            } else {
                                DeviceState newState = new DeviceState(event.getId(), deviceStatus, event.getOccurrenceTime());
                                DeviceState oldState = deviceState.value();
                                printDeviceStatusChange(newState, oldState, event);
                                deviceState.update(newState);
                            }
                        }
                    }
                })
                .print();

        env.execute();
    }

    private static Properties createKafkaProperties() {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
        // kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserClickedEventDeserializationSchema.class.getCanonicalName());
        return kafkaProps;
    }

    private static Integer getDeviceStatus(DeviceDataEvent event) {
        if (event.getPwrStat() == 0) {
            return 0;
        } else if (event.getPwrStat() == 1 && event.getWorkingStat() == 1) {
            return 1;
        } else if (event.getPwrStat() == 1 && event.getWorkingStat() == 0) {
            return 2;
        }

        return null;
    }

    private static void printDeviceStatusChange(DeviceState newState, DeviceState oldState, DeviceDataEvent event) {
        if (oldState.getStatus() == 0 && (newState.getStatus() == 1 || newState.getStatus() == 2)) {
            System.out.println("设备开机。相关事件：" + event.toString());
        } else if ((oldState.getStatus() == 1 || oldState.getStatus() == 2) && newState.getStatus() == 0) {
            System.out.println("设备关机。相关事件：" + event.toString());
        } else if (oldState.getStatus() == 1 && newState.getStatus() == 2) {
            System.out.println("设备开始待机。相关事件：" + event.toString());
        } else if (oldState.getStatus() == 2 && newState.getStatus() == 1) {
            System.out.println("设备开始工作。相关事件：" + event.toString());
        }
    }
}
